#include "ConnectionManager.h"
#include "commandconfig.h"
#include "utils.h"
#include "pbmessageconverter.h"
#include "timeoutexception.h"
//cms
#include "activemq/commands/ActiveMQTopic.h"
#include "activemq/commands/ActiveMQQueue.h"
#include "activemq/commands/ActiveMQTempQueue.h"
#include "activemq/commands/ActiveMQTempTopic.h"
#include "activemq/cmsutil/MessageCreator.h"
#include "decaf/util/UUID.h"

//qt
#include <QString>


using namespace activemq::core;
using namespace decaf::util::concurrent;
using namespace decaf::util;
using namespace decaf::lang;
using namespace cms;
using namespace std;

ConnectionManager::ConnectionManager(const std::string& name,cms::ConnectionFactory* factory
									 , const std::string& userName , const std::string& password ,const std::string& clientId )
:name(name),factory(factory), connection(NULL), userName(userName), password(password), clientId(clientId),
converter(new PbMessageConverter()), 
log( Log4Qt::Logger::logger("poc.ConnectionManager"))
{

}

ConnectionManager::~ConnectionManager(void)
{
	cleanup();
}

void ConnectionManager::close() {
	this->cleanup();
}

void ConnectionManager::open()
{

	if(converter == NULL)
	{
		converter = new PbMessageConverter();
	}
	// Create a Connection
	connection = factory->createConnection(userName, password, clientId);
	connection->start();
	connection->setExceptionListener(this);
	log->info("open the connection for uri %1", name.c_str());
	cmsTemplate = new JmsTemplate(connection, factory);
	cmsTemplate->setDeliveryPersistent(false);
	//cmsTemplate->setConnectionFactory(factory);

}

void ConnectionManager::send(CommandConfig* cmdCfg, google::protobuf::Message* msg) 
{

	std::string destinationName = cmdCfg->destinationName;
	std::string threadIdStr = Long::toString( Thread::getId() );
	log->info("sending message using thread id #%1. message: %2",  threadIdStr.c_str(), msg->DebugString().c_str());

	auto_ptr< Destination> destination ( createDestination(destinationName));

	class CreateBytesMessage : public activemq::cmsutil::MessageCreator
	{
		google::protobuf::Message* msg;
		IMessageConverter* inConveter;
	public:
		CreateBytesMessage(IMessageConverter* aConverter, google::protobuf::Message* msg)
			:inConveter(aConverter), msg(msg){}

		virtual cms::Message* createMessage( cms::Session* session )
		{

			return inConveter->toCmsMessage(msg);
		}
	};
	CreateBytesMessage func(converter, msg);
	cmsTemplate->send(destination.get(), &func);
}

google::protobuf::Message* ConnectionManager::call(CommandConfig* cmdCfg, google::protobuf::Message* msg) 
{

	std::string destinationName = cmdCfg->destinationName;
	std::string threadIdStr = Long::toString( Thread::getId() );
	log->info("call %1 message using thread id #%2. message: %3"
		, QString(cmdCfg->name.c_str()), QString( threadIdStr.c_str()), QString( msg->DebugString().c_str()));

	auto_ptr< Destination> destination ( createDestination(destinationName));
	auto_ptr<MessageHolder> holder( new MessageHolder(cmdCfg->timeout * 1000));
	std::string uuid = decaf::util::UUID::randomUUID().toString();
	cachedMsg.put(uuid, holder.get());

	class CreateBytesMessage : public activemq::cmsutil::MessageCreator
	{
		google::protobuf::Message* msg;
		IMessageConverter* inConveter;
		std::string uuid;
	public:
		CreateBytesMessage(IMessageConverter* aConverter, google::protobuf::Message* msg, const std::string& uuid)
			:inConveter(aConverter), msg(msg),uuid(uuid){}

		virtual cms::Message* createMessage( cms::Session* session )
		{

			 cms::Message* cmsMsg = inConveter->toCmsMessage(msg);
			 cmsMsg->setCMSCorrelationID(uuid);
			 return cmsMsg;
		}
	};
	CreateBytesMessage func(converter, msg, uuid);
	cmsTemplate->send(destination.get(), &func);
	holder->wait();
	cachedMsg.remove(uuid);
	if(holder->isCompleted())
	{
		return holder->popData();
	}
	else
	{
		throw TimeoutException();
	}

}

//void ConnectionManager::send(const std::string& destinationName,std::string& msg) 
//{
//
//
//	string threadIdStr = Long::toString( Thread::getId() );
//	log->info("sending message using thread id #%1. message: %2",  threadIdStr.c_str(), msg.c_str());
//
//	auto_ptr< Destination> destination ( createDestination(destinationName));
//
//	class CreateTextMessage : public activemq::cmsutil::MessageCreator
//	{
//		std::string& msg;
//	public:
//		CreateTextMessage(std::string& msg):msg(msg){}
//		virtual cms::Message* createMessage( cms::Session* session )
//		{
//			TextMessage* message = session->createTextMessage( msg );
//			return message;
//		}
//	};
//	CreateTextMessage func(msg);
//	cmsTemplate->send(destination.get(), &func);
//}

cms::Destination* ConnectionManager::createDestination(const std::string& destinationName)
{

	if(destinationName.find("topic:") == 0)
	{
		std::string name = destinationName.substr(6);
		return new activemq::commands::ActiveMQTopic(name);
	}
	else if(destinationName.find("queue:") == 0)
	{
		std::string name = destinationName.substr(6);
		return new activemq::commands::ActiveMQTopic(name);
	}
	else if(destinationName.find("temptopic:") == 0)
	{
		std::string name = destinationName.substr(10);
		return new activemq::commands::ActiveMQTempTopic();
	}
	else if(destinationName.find("tempqueue:") == 0)
	{
		std::string name = destinationName.substr(10);
		return new activemq::commands::ActiveMQTempQueue();
	}
	else 
	{
		return new activemq::commands::ActiveMQQueue(destinationName);
	}
}

void ConnectionManager::listen(CommandConfig* cmdCfg)
{
	std::string destinationName = cmdCfg->destinationName;
	std::vector<CommonConsumer*>::iterator consumer = findConsumer(cmdCfg->name);

	if(consumer == consumers.end())
	{
		CommonConsumer* consumer = new CommonConsumer(this->connection, cmdCfg, this);
		consumer->init();
		consumers.push_back(consumer);
		log->info("listenered %1", cmdCfg->name.c_str());
	}
}

std::vector<CommonConsumer*>::iterator ConnectionManager::findConsumer(const std::string &cmdName)
{
	class NameFinder
	{
	private:
		std::string name;
	public:
		NameFinder(const std::string& name):name(name){}
		bool operator()(CommonConsumer* consumer)
		{
			return consumer->getCfg()->name == name;
		}
	};
	std::vector<CommonConsumer*>::iterator result = std::find_if(consumers.begin(), consumers.end(), NameFinder(cmdName));
	return result;
}

void ConnectionManager::stopListen(const std::string &cmdName)
{
	std::vector<CommonConsumer*>::iterator result = findConsumer(cmdName);
	if(result != consumers.end())
	{
		CommonConsumer* consumer = *result;
		consumer->decreaseRef();
		if(consumer->getRefCount() == 0)
		{
			consumers.erase(result);
			delete consumer;
			log->info("stoped listener %1", cmdName.c_str());
		}

	}
}

void ConnectionManager::cleanup(){
	log->info("clean connection manager.");

	if(converter != NULL)
	{
		delete converter;
	}
	std::vector<MessageHolder*> values = cachedMsg.values();
	for(std::vector<MessageHolder*>::const_iterator it = values.begin(); it != values.end(); it++)
	{
		(*it)->notifyAll();
	}
	cachedMsg.clear(); // just notify thread, the creating holder thread  be responsible for the deleting holder object.

	class DeleteCommonConsumer
	{
		Log4Qt::Logger* log;
	public:
		DeleteCommonConsumer(Log4Qt::Logger * logger):log(logger){}
		void operator()(CommonConsumer* consumer)
		{
			log->info(("Deleting common consumer. " + consumer->getCfg()->name).c_str()); 
			try
			{
				if( consumer != NULL ) delete consumer;
			}catch ( CMSException& e ) { log->error(("Error when deleting consumer. " + e.getStackTraceString()).c_str()); }


		}
	};

	std::for_each(consumers.begin(), consumers.end(), DeleteCommonConsumer(log));
	// Close open resources.

	if(cmsTemplate != NULL)
	{
		delete cmsTemplate;
	}
	try{
		if( connection != NULL ) delete connection;
	}catch ( CMSException& e ) { log->error(("Error when releasing connection. " + e.getStackTraceString()).c_str()); }
	catch(decaf::lang::exceptions::RuntimeException& re) { log->error(("Error when releasing connection. " + re.getStackTraceString()).c_str());}
	connection = NULL;

}

// Called from the consumer since this class is a registered MessageListener.
void ConnectionManager::onMessage( const Message* message ) throw()
{

	try
	{
		void* voidMsg = this->converter->toObject(message);
		std::auto_ptr<google::protobuf::Message> pbMsg (static_cast<google::protobuf::Message*>(voidMsg));

		string threadIdStr = Long::toString( Thread::getId() );
		log->info("received message using thread id #%1. message: %2", threadIdStr.c_str(), pbMsg->DebugString().c_str());
		std::string uuid = message->getCMSCorrelationID();
		if(cachedMsg.containsKey(uuid))
		{
			
			MessageHolder* holder = cachedMsg.get(uuid);
			holder->putData(pbMsg.release()); // the data will be returned to caller, it should not be deleted.
			holder->notifyAll();
			return;
		}
		//for event, message objects should be deleted by this method.
		std::auto_ptr<MessageArgs> args (new MessageArgs(pbMsg.get(), message));
		fireEvent(args.get());

	} catch (CMSException& e) {
		log->error(("Error when handling message. " + e.getStackTraceString()).c_str());
	}


}


void ConnectionManager::onException( const CMSException& ex AMQCPP_UNUSED) {
	printf("CMS Exception occurred.  Shutting down client.\n");
	ex.printStackTrace();
}